package io.kiki.sba.registry.server.clientInterface.strategy.impl;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.common.model.store.Watcher;
import io.kiki.sba.registry.core.constants.EventTypeConstants;
import io.kiki.sba.registry.core.model.ConfiguratorRegister;
import io.kiki.sba.registry.core.model.RegisterResponse;
import io.kiki.sba.registry.remoting.netty.BoltUtil;
import io.kiki.sba.registry.server.clientInterface.converter.SubscriberConverter;
import io.kiki.sba.registry.server.clientInterface.registry.Registry;
import io.kiki.sba.registry.server.clientInterface.strategy.WatcherHandlerStrategy;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import static io.kiki.sba.registry.common.model.constants.ValueConstants.DEFAULT_INSTANCE_ID;


public class DefaultWatcherHandlerStrategy implements WatcherHandlerStrategy {
    private static final Logger logger = LoggerFactory.getLogger(DefaultWatcherHandlerStrategy.class);

    @Autowired
    protected Registry sessionRegistry;

    @Override
    public void handleConfiguratorRegister(Channel channel, ConfiguratorRegister configuratorRegister, RegisterResponse registerResponse) {
        Watcher watcher = null;
        try {
            configuratorRegister.setIp(channel.getRemoteAddress().getAddress().getHostAddress());
            configuratorRegister.setPort(channel.getRemoteAddress().getPort());

            if (StringUtils.isBlank(configuratorRegister.getInstanceId())) {
                configuratorRegister.setInstanceId(DEFAULT_INSTANCE_ID);
            }

            watcher = SubscriberConverter.convert(configuratorRegister);
            watcher.setProcessId(channel.getRemoteAddress().getHostName() + ":" + channel.getRemoteAddress().getPort());

            handle(watcher, channel, configuratorRegister, registerResponse);
        } catch (Throwable e) {
            handleError(configuratorRegister, watcher, registerResponse, e);
        }
    }

    protected void handle(Watcher watcher, Channel channel, ConfiguratorRegister register, RegisterResponse registerResponse) {
        watcher.setClientUrl(new Url(channel.getRemoteAddress(), BoltUtil.getBoltCustomSerializer(channel)));
        watcher.setServerUrl(new Url(channel.getLocalAddress()));

        final String eventType = register.getEventType();
        if (EventTypeConstants.REGISTER.equals(eventType)) {
            sessionRegistry.register(channel, watcher);
        } else
            if (EventTypeConstants.UNREGISTER.equals(eventType)) {
                sessionRegistry.unRegister(watcher);
            } else {
                logger.warn("unsupported watch.eventType:{}", eventType);
            }
        registerResponse.setVersion(register.getVersion());
        registerResponse.setRegistId(register.getRegistId());
        registerResponse.setSuccess(true);
        registerResponse.setMessage("ConfiguratorRegister register success!");
        log(true, register, watcher);
    }

    private void log(boolean success, ConfiguratorRegister register, Watcher watcher) {
        // [Y|N],[R|U|N],app,zone,dataInfoId,registerId,clientVersion,clientIp,clientPort
        Metrics.Access.watCount(success);
        logger.info("{},{},{},{},{},{},{},{},{},{},{},attrs={}", success ? 'Y' : 'N', EventTypeConstants.getEventTypeFlag(register.getEventType()), register.getAppName(), register.getZone(), register.getDataId(), register.getGroup(), register.getInstanceId(), register.getRegistId(), watcher == null ? "" : watcher.getClientVersion(), register.getIp(), register.getPort(), watcher == null ? "0" : watcher.attributesSize());
    }

    protected void handleError(ConfiguratorRegister register, Watcher watcher, RegisterResponse registerResponse, Throwable e) {
        log(false, register, watcher);
        RegisterLogs.logError(register, "Watcher", registerResponse, e);
    }
}
